package com.wpp.client;

import com.wpp.boot.ConsumerBootstrap;
import com.wpp.constant.Constant;
import com.wpp.handler.UserClientHandler;
import com.wpp.protocol.RpcRequest;
import com.wpp.register.loaderbalance.LoaderBalance;
import com.wpp.register.loaderbalance.MinTimeLoaderBalance;
import com.wpp.register.loaderbalance.RandomLoaderBalance;
import com.wpp.register.ServerInfo;
import com.wpp.register.ZkRegisterCenter;
import com.wpp.register.converter.ServerInfoConvert;
import com.wpp.register.metric.ResponseTimeMetricHolder;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.utils.ZKPaths;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author by wpp25
 * @Classname RPCConsumer
 * @Description  消费者
 * @Date 2020/11/25 21:06
 */
@Component
public class RPCConsumer implements InitializingBean {


    private ZkRegisterCenter registerCenter;

    // 创建一个线程池对象
    private ExecutorService executorService;

    private ConsumerBootstrap consumerBootstrap;
    private LoaderBalance loaderBalance;

    /**
     * 服务连接缓存列表
     * key : ip + port
     * value : 当前连接 服务列表
     */
    private final Map<String,Set<ServerInfo>> SERVER_INFO_MAP = new ConcurrentHashMap<>();

    /**
     *  添加监听 监听提供该service  的 服务列表
     * @param serviceName
     * @throws Exception
     */
    private void addWatchListener(String serviceName) throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(registerCenter.getCuratorFramework(),
                new StringBuilder(Constant.ZkConfig.REGISTER_ROOT).append(Constant.ZkConfig.SPILT).append(serviceName).toString(),
                true);
        PathChildrenCacheListener pathChildrenCacheListener = (curatorFramework, event) -> {
            String serviceServerNote = ZKPaths.getNodeFromPath(event.getData().getPath());
            ServerInfo changeServer = ServerInfoConvert.convert(serviceServerNote);
            switch (event.getType()) {
                // 新增服务
                case CHILD_ADDED: {
                    // 新增服务提供者的时候  将 新进来的服务 放入到本地缓存起来 同时 和当前服务 建立连接
                    SERVER_INFO_MAP.get(serviceName).add(changeServer);
                    consumerBootstrap.initClient(changeServer);
//                    SERVER_INFO_MAP.get(serviceName).add(ZKPaths.getNodeFromPath(serviceServerNote));
                    System.out.println("Child Node added: " + serviceServerNote);

                    break;
                }
                case CHILD_REMOVED: {
                    // 服务提供者下线的时候 将 新下线的服务 放入到本地缓存起来 同时 和当前服务 建立连接
                    Set<ServerInfo> serverInfoSet = SERVER_INFO_MAP.get(serviceName);
                    for (ServerInfo serverInfo : serverInfoSet) {
                        if (changeServer.getIp().equals(serverInfo.getIp()) && changeServer.getPort() == serverInfo.getPort()) {
                            if (serverInfo.getChannel() != null) {
                                // 服务下线 关闭连接
                                serverInfo.getChannel().close().sync();
                            }
                            serverInfoSet.remove(serverInfo);
                            break;
                        }
                    }
                    // 服务下线 将 该服务对应的响应时间监控从缓存中移除
                    ResponseTimeMetricHolder.getInstance().getTimeMetricMap().remove(serviceServerNote);
                    System.out.println("Child Node removed: " + serviceServerNote);
                    break;
                }
                default:
                    break;
            }
        };
        // 重新建立监听
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        pathChildrenCache.start();

    }
    // jdk 动态代理创建对象
    /**
     *
     * @param serviceClass 接口类型，根据哪个接口生成子类代理对象
     * @return
     */
    public Object createProxy(Class<?> serviceClass) {

        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        String serviceName = serviceClass.getName();

                        if (!SERVER_INFO_MAP.containsKey(serviceName)) {
                            // 加入缓存中 没有当前服务对应的列表 则 将该服务列表加载到 缓存中
                            SERVER_INFO_MAP.put(serviceName,registerCenter.getServerList(serviceName));
                            addWatchListener(serviceName);
                        }
                        Set<ServerInfo> serverInfoSet = SERVER_INFO_MAP.get(serviceName);
                        if (CollectionUtils.isEmpty(serverInfoSet)) {
                            throw  new IllegalArgumentException("no server provide" + ":" + serviceName);
                        }


                        ServerInfo serverInfo = loaderBalance.balance(new ArrayList<>(serverInfoSet));
                        if (serverInfo.getChannel() == null) {
                            consumerBootstrap.initClient(serverInfo);
                        }
                        UserClientHandler userClientHandler = (UserClientHandler)serverInfo.getChannel().pipeline().last();

                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setRequestId(UUID.randomUUID().toString());
                        rpcRequest.setClassName(serviceName);
                        rpcRequest.setMethodName(method.getName());
                        rpcRequest.setParameterTypes(method.getParameterTypes());
                        rpcRequest.setParameters(args);
                        rpcRequest.setServerAddress(new StringBuilder(serverInfo.getIp()).append(":").append(serverInfo.getPort()).toString());
                        userClientHandler.setRpcRequest(rpcRequest);

                         // 给UserClientHandler设置param参数
                        Object o = executorService.submit(userClientHandler).get();

                        return o;
                    }

                }

        );
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        this.registerCenter =  new ZkRegisterCenter();
        this.executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        this.consumerBootstrap = new ConsumerBootstrap(registerCenter);
        this.loaderBalance = new MinTimeLoaderBalance();

    }
}
